package com.hong.tubemq.consumer;

import com.alibaba.fastjson.JSONObject;
import org.apache.tubemq.client.config.ConsumerConfig;
import org.apache.tubemq.client.consumer.ConsumePosition;
import org.apache.tubemq.client.consumer.ConsumerResult;
import org.apache.tubemq.client.consumer.PullMessageConsumer;
import org.apache.tubemq.client.factory.MessageSessionFactory;
import org.apache.tubemq.client.factory.TubeSingleSessionFactory;
import org.apache.tubemq.corebase.Message;
import org.apache.tubemq.corebase.utils.ThreadUtils;

import java.util.List;
/**
 *
 * 功能描述: 通过pull拉取消息
 *
 * @param: 
 * @return: 
 * @auther: csh
 * @date: 2021/5/16 17:51
 */
public class PullConsumerExample {

     public static void main(String[] args) throws Throwable {
         //服务地址
         final String masterHostAndPort = "localhost:8000";
         //topic
         final String topic = "java_tubemq";
         //消费组
         final String group = "hong-group1";
         final ConsumerConfig consumerConfig = new ConsumerConfig(masterHostAndPort, group);
         consumerConfig.setConsumePosition(ConsumePosition.CONSUMER_FROM_LATEST_OFFSET);
         final MessageSessionFactory messageSessionFactory = new TubeSingleSessionFactory(consumerConfig);
         final PullMessageConsumer messagePullConsumer = messageSessionFactory.createPullConsumer(consumerConfig);
         messagePullConsumer.subscribe(topic, null);
         messagePullConsumer.completeSubscribe();
         // wait for client to join the exact consumer queue that consumer group allocated
         while (!messagePullConsumer.isPartitionsReady(1000)) {
             ThreadUtils.sleep(1000);
         }
         while (true) {
             ConsumerResult result = messagePullConsumer.getMessage();
             if (result.isSuccess()) {
                 List<Message> messageList = result.getMessageList();
                 for (Message message : messageList) {
                     System.out.println("接收到的消息: " + JSONObject.toJSONString(message)+"内容是:"+new String(message.getData()));
                 }
                 messagePullConsumer.confirmConsume(result.getConfirmContext(), true);
             } else {
                 if (result.getErrCode() == 400) {
                     ThreadUtils.sleep(100);
                 } else {
                     if (result.getErrCode() != 404) {
                         System.out.println(String.format("Receive messages errorCode is %d, Error message is %s", result.getErrCode(), result.getErrMsg()));
                     }
                 }
             }
         }
     }   

 }